package com.py.custom.ability.kafka.producer;

import com.py.custom.ability.kafka.producer.base.DefaultProducerProcessor;
import com.py.custom.ability.kafka.producer.base.ProducerProcessor;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;

import java.util.Properties;

/**
 * kafka生产者的代理：用来代理用户的发送
 * custom.ckafka.producer[0].producerName=producer0
 * custom.ckafka.producer[0].brokerServers=集群信息
 * custom.ckafka.producer[0].topicName=消息主题
 * custom.ckafka.producer[0].hasTransaction=false
 */
public class CKafkaProducerProxy<T> implements FactoryBean<T>, InitializingBean {

    /**
     * 生产者
     */
    private ProducerProcessor producerProcessor;

    /**
     * 生产者的配置信息
     */
    private CKafkaProducerProperties cKafkaProducerProperties;


    public CKafkaProducerProperties getcKafkaProducerProperties() {
        return cKafkaProducerProperties;
    }

    public void setcKafkaProducerProperties(CKafkaProducerProperties cKafkaProducerProperties) {
        this.cKafkaProducerProperties = cKafkaProducerProperties;
    }

    @Override
    public T getObject() throws Exception {
        if (cKafkaProducerProperties == null) {
            return null;
        }
        return (T) producerProcessor;
    }

    @Override
    public Class<?> getObjectType() {
        return ProducerProcessor.class;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cKafkaProducerProperties.getBrokerServers());
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // todo CKafkaProducerProperties配置信息待完善
        producerProcessor = new DefaultProducerProcessor(cKafkaProducerProperties.getTopicName(), properties);
    }

    public void close() throws Exception {
        producerProcessor.close();
    }
}
